<html>
 <head>
  <meta charset="UTF-8">
 </head>
 <body>
  <h1 data-lake-id="JDYwv" id="JDYwv"><span data-lake-id="uf4272a08" id="uf4272a08">典型回答</span></h1>
  <p data-lake-id="u41a6de77" id="u41a6de77"><br></p>
  <p data-lake-id="ub83ddcc5" id="ub83ddcc5"><span data-lake-id="u82365178" id="u82365178">在Kafka中，想要提升吞吐量，有效的手段就是增加消费者实例以及增加更多的partition，但是如果不能增加消费者实例，也不能新增更多的partition，那么如何通过技术手段提高吞吐量呢？</span></p>
  <p data-lake-id="uaf8309d2" id="uaf8309d2"><span data-lake-id="u0d646de6" id="u0d646de6">​</span><br></p>
  <p data-lake-id="ud0d0f739" id="ud0d0f739"><span data-lake-id="u79777fa0" id="u79777fa0">主要有以下几个方面可以做：</span></p>
  <p data-lake-id="uc1ec41b3" id="uc1ec41b3"><br></p>
  <p data-lake-id="u4177392e" id="u4177392e"><span data-lake-id="u5fae3b87" id="u5fae3b87">1、异步消费</span></p>
  <p data-lake-id="u6d427ec5" id="u6d427ec5"><span data-lake-id="u5070711a" id="u5070711a">2、增加消费的线程数</span></p>
  <p data-lake-id="u50a7e55f" id="u50a7e55f"><span data-lake-id="uf99d7970" id="uf99d7970">4、消息压缩或组合</span></p>
  <p data-lake-id="u3451dd56" id="u3451dd56"><span data-lake-id="u0125a09b" id="u0125a09b">4、调整Kafka参数</span></p>
  <p data-lake-id="u848c2521" id="u848c2521"><span data-lake-id="u38e68cd2" id="u38e68cd2">​</span><br></p>
  <h3 data-lake-id="nVYkI" id="nVYkI"><span data-lake-id="udeac3b91" id="udeac3b91">异步消费</span></h3>
  <p data-lake-id="ud9da82a6" id="ud9da82a6"><br></p>
  <p data-lake-id="u4d6c275c" id="u4d6c275c"><span data-lake-id="ucca2fb91" id="ucca2fb91">想要提升消费者的消费速度，我们可以快速的处理消息，有一个办法就是在接收到消息之后，先本地落库，然后落库成功之后就直接提交偏移量。</span></p>
  <p data-lake-id="u6a2b33a4" id="u6a2b33a4"><span data-lake-id="uc15f0671" id="uc15f0671">​</span><br></p>
  <p data-lake-id="u9e613d34" id="u9e613d34"><span data-lake-id="ua9caec1e" id="ua9caec1e">然后在用异步的方式消费这些任务。为了提升速度，我们可以在消费消息的线程中，本地消息保存成功之后，直接起一个异步线程来处理这个消息，如果成功，则直接把消息删除掉，如果失败，那么依赖本地消息重试。</span></p>
  <p data-lake-id="ufcb7622d" id="ufcb7622d"><span data-lake-id="u7217178e" id="u7217178e">​</span><br></p>
  <p data-lake-id="uc5786e3f" id="uc5786e3f"><span data-lake-id="ued78b06c" id="ued78b06c">那有人就会问了，这样做，还有消息中间件有意义吗？</span></p>
  <p data-lake-id="u8d5bfb24" id="u8d5bfb24"><span data-lake-id="u49a959fc" id="u49a959fc">​</span><br></p>
  <p data-lake-id="u7c4ea700" id="u7c4ea700"><span data-lake-id="u13994df3" id="u13994df3">当然有了，消息中间件的目的有很多个，比如解耦、异步、削峰填谷。我们在消费者端，接到消息后本地存下来再执行，不影响我们依然做到了解耦、异步和削峰填谷。</span></p>
  <p data-lake-id="uaf4edfc9" id="uaf4edfc9"><br></p>
  <h3 data-lake-id="kE4f2" id="kE4f2"><span data-lake-id="uadf74436" id="uadf74436">多线程消费</span></h3>
  <p data-lake-id="u26162e0b" id="u26162e0b"><br></p>
  <p data-lake-id="u3527cab8" id="u3527cab8"><span data-lake-id="u808e333f" id="u808e333f">如果不想用本地落库的方式，那么也可以直接用多线程来执行。</span></p>
  <p data-lake-id="u0799b855" id="u0799b855"><span data-lake-id="u605d804d" id="u605d804d">​</span><br></p>
  <p data-lake-id="uc5676cfd" id="uc5676cfd"><span data-lake-id="u9efdc37c" id="u9efdc37c">我们可以通过配置，让kafka一次性多拉取一些消息，在多个消息都拉取到之后，通过异步线程池的方式来并发执行。</span></p>
  <p data-lake-id="u85b5dce7" id="u85b5dce7"><span data-lake-id="u27f7be0e" id="u27f7be0e">​</span><br></p>
  <p data-lake-id="ua82d6c81" id="ua82d6c81"><span data-lake-id="u88fed8ed" id="u88fed8ed">这种方案可以借助多线程来并发消费一批消息，从而提升并发度，提高吞吐量。但是这个方案存在一个比较大的问题，那就是异步线程多个消息的处理时间不同，可能会导致偏移量的提交并不能按照顺序，那么这个过程可能就会存在消息的丢失和重复消费的情况。</span></p>
  <p data-lake-id="u4e032443" id="u4e032443"><br></p>
  <p data-lake-id="uffc9aec1" id="uffc9aec1"><span data-lake-id="ueb059c55" id="ueb059c55">那么，有什么办法可以解决这个问题呢？</span></p>
  <p data-lake-id="uc7279406" id="uc7279406"><span data-lake-id="u1e074b28" id="u1e074b28">​</span><br></p>
  <p data-lake-id="u1bab5248" id="u1bab5248"><strong><span data-lake-id="u21ec4f36" id="u21ec4f36">第一种方案</span></strong><span data-lake-id="u5ea62a05" id="u5ea62a05">，我们可以把</span><strong><span data-lake-id="ud57248e1" id="ud57248e1">多个线程编排起来</span></strong><span data-lake-id="u1ae92e20" id="u1ae92e20">执行，当所有的线程都消费成功之后，把这其中最大的偏移量提交了就行了。比如我们可以使用</span><span data-lake-id="uf3ff3266" id="uf3ff3266" style="color: rgb(38, 38, 38)">CompletableFuture轻松的实现这个功能，CompletableFuture即自带线程池，又支持任务分片，非常适合这种场景，如：</span></p>
  <p data-lake-id="u68a28531" id="u68a28531"><span data-lake-id="u2ffbd9bc" id="u2ffbd9bc" style="color: rgb(38, 38, 38)">​</span><br></p>
  <pre lang="java"><code>
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;

public class KafkaConsumerExample {

    private KafkaConsumer&lt;String, String&gt; consumer;
    private List&lt;ConsumerRecord&lt;String, String&gt;&gt; records = new ArrayList&lt;&gt;();
    
    public void consume(int batchSize) {
        // 从 Kafka 中拉取 batchSize 条消息
        records = consumer.poll(batchSize);
        
        // 使用 CompletableFuture 并发执行消息处理
        CompletableFuture&lt;Void&gt; allFutures = CompletableFuture.allOf(records.stream()
        .map(detail -&gt; CompletableFuture.supplyAsync(() -&gt; {
            // 处理消息逻辑
            // ...
						//如果消费失败，直接throw Exception
            return null;
        })).toArray(CompletableFuture[]::new));        

      	// 等待所有任务执行完毕后，提交最大的偏移量
        allFutures.whenComplete((v,e) -&gt; {
        	if(e == null){
          	//未发生异常，则提交最大的偏移量
          	long maxOffset = records.stream().mapToLong(ConsumerRecord::offset).max().orElse(-1L);
            consumer.commitSync(Collections.singletonMap(consumer.assignment().iterator().next(), new OffsetAndMetadata(maxOffset + 1)));
          }else{
          	//有消息消费失败，执行失败处理逻辑
          }
        });
    }
}

</code></pre>
  <p data-lake-id="u82cdafd1" id="u82cdafd1"><span data-lake-id="uc56e917f" id="uc56e917f">​</span><br></p>
  <p data-lake-id="u165e0e68" id="u165e0e68"><span data-lake-id="uf8214662" id="uf8214662">这样可能存在一个问题，就是如果其中某个消息执行失败了，就会导致没办法正确提交偏移量，导致很多消息会重复消费。很多时候如果我们做好了幂等，重复消费的问题理论上可以忽略，但是如果要解决的话，我们也有一些办法。</span></p>
  <p data-lake-id="u802e64ca" id="u802e64ca"><span data-lake-id="ud9c61b5d" id="ud9c61b5d">​</span><br></p>
  <p data-lake-id="u4af52c8f" id="u4af52c8f"><span data-lake-id="ud2911395" id="ud2911395">比如如果有失败的，那么就想办法把失败的这个消息之前的偏移量都提交了，之后的不提交。可以减少一些重复消费的消息的数量。或者我们干脆把失败的消息落库，通过扫表的方式做重试。</span></p>
  <p data-lake-id="ua02ad560" id="ua02ad560"><span data-lake-id="u3c453a0f" id="u3c453a0f">​</span><br></p>
  <p data-lake-id="ua7991352" id="ua7991352"><strong><span data-lake-id="u04c295a2" id="u04c295a2">第二种方案，</span></strong><span data-lake-id="u20a3a712" id="u20a3a712">因为题目中说了我们只有一个partition和一个消费者，所以正常情况下，偏移量应该是连续的，所以我我们在一次性拉取多个消息之后，可以知道这一批消息的最大偏移量和最小偏移量是多少。这时候可以记录下来，并且在每一个消息消费成功之后，也把他的偏移量记录下来。</span></p>
  <p data-lake-id="u67db72cc" id="u67db72cc"><span data-lake-id="u86f5e71f" id="u86f5e71f">​</span><br></p>
  <p data-lake-id="u585fadbb" id="u585fadbb"><span data-lake-id="u8781203e" id="u8781203e">可以用数据库或者redis，如果是这种只有单实例的情况，其使用</span><strong><span data-lake-id="u9a4450c9" id="u9a4450c9">本地缓存</span></strong><span data-lake-id="uebb57757" id="uebb57757">也可以，既简单，又方便，又不会有分布式一致性问题，唯一的缺点就是机器重启可能会丢，但是这个问题不是很大，大不了重复消费呗，做好幂等就行了。</span></p>
  <p data-lake-id="u764d9236" id="u764d9236"><span data-lake-id="ufb887b47" id="ufb887b47">​</span><br></p>
  <p data-lake-id="u3e43436c" id="u3e43436c"><span data-lake-id="u30c393f1" id="u30c393f1">这样在一定的时间延迟之后，我们就能得到一些已经处理过的消息的偏移量。比如100,101,102,104,105。这时候我们就从小到大，一直找到第一个不连续的偏移量，把他提交了，即102。这种方案需要借助第三方的一个存储。</span></p>
  <p data-lake-id="u854fdf23" id="u854fdf23"><span data-lake-id="u8e04138b" id="u8e04138b">​</span><br></p>
  <p data-lake-id="u6c38dced" id="u6c38dced"><span data-lake-id="u6a16445e" id="u6a16445e">借助这个方案，我们可以解决消息丢失的问题，因为103这个消息没成功，但是偏移量我们提交的是102的这个，所以这时候103的消息会重发，103,104,105,106.107会在下一次被拉取到。这时候我们还是多线程消费，同时判断104 和 105已经消费过了，就不需要重新消费了，只需要消费另外几个就行了。</span></p>
  <p data-lake-id="u2d53dba1" id="u2d53dba1"><span data-lake-id="u5f633030" id="u5f633030">​</span><br></p>
  <h3 data-lake-id="jFSb7" id="jFSb7"><span data-lake-id="uaf1d0e2c" id="uaf1d0e2c">消息压缩和组合</span></h3>
  <p data-lake-id="u733c71ca" id="u733c71ca"><br></p>
  <p data-lake-id="udd879ef9" id="udd879ef9"><span data-lake-id="u23dfacdd" id="u23dfacdd">Kafka支持多种压缩算法，如gzip、snappy、lz4等，我们可以适当的采用合适的算法将消息进行压缩，消息内容小就可以减少网络传输的数据量，从而提升整体的吞吐量。</span></p>
  <p data-lake-id="u2de833d9" id="u2de833d9"><span data-lake-id="u3470fe13" id="u3470fe13">​</span><br></p>
  <p data-lake-id="u3f48acab" id="u3f48acab"><span data-lake-id="u7bb3cb3c" id="u7bb3cb3c">除了压缩之外，组合也是一个好的办法，如果消息的生产者可以配合我们一起做改造的话，我们可以要求生产者在发送消息的时候，把多条消息组合到一起作为一个大消息发送。然后消费者收到之后再把消息解组之后消费。这样也能降低消息的数量，从而提升系统的吞吐量。</span></p>
  <p data-lake-id="u65747af4" id="u65747af4"><span data-lake-id="ue90db3a7" id="ue90db3a7">​</span><br></p>
  <h3 data-lake-id="awiSt" id="awiSt"><span data-lake-id="ue18c7880" id="ue18c7880">调整kafka参数</span></h3>
  <p data-lake-id="u907230d3" id="u907230d3"><br></p>
  <p data-lake-id="uf179c550" id="uf179c550"><span data-lake-id="u5831de78" id="u5831de78">前面我们提到过的异步消费、多线程消费、压缩等，都是有一些前提的，比如要把offset的提交方式改为手动提交，每次拉取消息的数量、压缩方式等等。</span></p>
  <p data-lake-id="u16052fae" id="u16052fae"><span data-lake-id="ue294f121" id="ue294f121">​</span><br></p>
  <pre lang="java"><code>
enable.auto.commit = false //手动提交

fetch.max.bytes = 104857600 //设置每次拉取请求的最大数据量为100MB
max.poll.records = 50 // 设置每次调用poll()方法最多拉取50条记录

compression.type = gzip //使用gzip进行消息压缩
</code></pre>
  <p data-lake-id="u977e50bf" id="u977e50bf"><span data-lake-id="u613bea1c" id="u613bea1c">​</span><br></p>
  <p data-lake-id="ub5589d93" id="ub5589d93"><span data-lake-id="uec56063b" id="uec56063b">除此以外，还有一些其他的参数调整合适也是可以提升吞吐量的，比较常用的有：</span></p>
  <p data-lake-id="ub4a48fd1" id="ub4a48fd1"><span data-lake-id="u4e1da567" id="u4e1da567">​</span><br></p>
  <pre lang="java"><code>
num.network.threads = 16 //Broker处理消息的最大线程数，可以设置成CPU数的2倍
num.io.threads = 16 // Broker处理磁盘IO的线程数，可以设置成CPU数的2倍

socket.send.buffer.bytes=65536 //发送缓冲区的大小
socket.receive.buffer.bytes=65536 //接收缓冲区的大小
</code></pre>
  <p data-lake-id="ua408caf6" id="ua408caf6"><span data-lake-id="u40ff053b" id="u40ff053b">​</span><br></p>
  <blockquote data-lake-id="u35e043c4" id="u35e043c4">
   <p data-lake-id="u819192cc" id="u819192cc"><span data-lake-id="u2e9245ef" id="u2e9245ef">num.network.threads 指定 Kafka Broker 用于处理网络请求的线程数，它主要负责处理来自客户端的请求、响应请求以及数据传输等。</span></p>
   <p data-lake-id="u7f877ee5" id="u7f877ee5"><span data-lake-id="ucbed2cd6" id="ucbed2cd6">num.io.threads 指定 Kafka Broker 用于处理磁盘 I/O 的线程数，它主要负责读写磁盘上的数据。</span></p>
   <p data-lake-id="uf72c52fa" id="uf72c52fa"><span data-lake-id="ucdba5443" id="ucdba5443">socket.send.buffer.bytes 控制发送缓冲区的大小，即客户端发送消息到 Kafka broker 时的缓冲区大小。</span></p>
   <p data-lake-id="ude07b8bb" id="ude07b8bb"><span data-lake-id="ue15a92fa" id="ue15a92fa">socket.receive.buffer.bytes 控制接收缓冲区的大小，即客户端从 Kafka broker 接收消息时的缓冲区大小。</span></p>
  </blockquote>
  <p data-lake-id="uaa1dba4a" id="uaa1dba4a"><span data-lake-id="uc409b910" id="uc409b910" class="lake-fontsize-12" style="color: rgb(55, 65, 81); background-color: rgb(247, 247, 248)">​</span><br></p>
  <p data-lake-id="u41142a24" id="u41142a24"><span data-lake-id="u9b3c78b3" id="u9b3c78b3">一般来说，参数的合理设置需要考虑硬件配置、Kafka Broker 的负载情况以及网络吞吐量等多个因素。</span></p>
 </body>
</html>